package com.snail.starter.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author: snail framework
 * @Date: 2023/1/31 下午4:00
 * @Version: 1.0.0
 */
@Slf4j
@Component
public class KafkaApplicationContext implements InitializingBean, DisposableBean {

    @Resource
    ApplicationContext applicationContext;

    @Resource
    Environment environment;

    @Value("${spring.application.name}")
    String applicationName;

    @Resource
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory;

    ConcurrentHashMap<String, ConcurrentMessageListenerContainer<String, String>> cache = new ConcurrentHashMap<>();

    @Override
    public void afterPropertiesSet() throws Exception {
        log.info("======================== Kafka Listener Start ========================");
        Map<String, KafkaListener> kafkaListenerMap = applicationContext.getBeansOfType(KafkaListener.class);
        if (kafkaListenerMap.isEmpty()) {
            return;
        }
        kafkaListenerMap.values().forEach(this::subscribe);
        log.info("======================== Kafka Listener Finish ========================");
    }

    public void subscribe(KafkaListener kafkaListener) {
        //topic
        String topic;
        if (!"production".equalsIgnoreCase(environment.getActiveProfiles()[0])) {
            topic = kafkaListener.topic() + "_" + environment.getActiveProfiles()[0];
        } else {
            topic = kafkaListener.topic();
        }
        ConcurrentMessageListenerContainer<String, String> container = kafkaListenerContainerFactory.createContainer(topic);
        container.setConcurrency(kafkaListener.concurrency());
        container.setupMessageListener(kafkaListener);
        //group
        String group;
        if (null == kafkaListener.group() || "".equals(kafkaListener.group())) {
            group =  applicationName;
        } else {
            group = kafkaListener.group();
        }
        container.getContainerProperties().setGroupId(group);
        cache.put(uniqueConsumer(topic, group), container);
        container.start();
        log.info("======================== Kafka subscribe topic:{},group:{}, concurrency:{} ========================", kafkaListener.topic(), group, kafkaListener.concurrency());
    }

    /**
     * 全部取消订阅
     */
    public void unsubscribeAll() {
        cache.forEach((uniqueConsumer, container) -> {
            container.stop();
            log.info("======================== Kafka unsubscribe, uniqueConsumer:{} ========================", uniqueConsumer);
        });
    }

    /**
     * 取消订阅单个topic的某个分组
     *
     * @param topic topic/主题
     * @param group 分组
     */
    public void unsubscribe(String topic, String group) {
        String uniqueConsumer = uniqueConsumer(topic, group);

        ConcurrentMessageListenerContainer<String, String> container = cache.get(uniqueConsumer);
        if (null != container) {
            container.stop();
            log.info("======================== unsubscribe uniqueConsumer:{} ========================", uniqueConsumer);
        }
    }

    /**
     * 所有的消息者都进行订阅
     *
     * @throws Exception
     */
    public void subscribeAll() throws Exception {
        afterPropertiesSet();
    }

    public static String uniqueConsumer(String topic, String group) {
        return topic + "|" + group;
    }

    @Override
    public void destroy() throws Exception {
        log.info("======================== all consumer start unsubscribe ！！！ ========================");
        unsubscribeAll();
        log.info("======================== all consumer finish unsubscribe ！！！ ========================");
    }
}